/*
 * Copyright (c) 2010 Vadim Zhukov <persgray@gmail.com>
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */


#include "common.h"

#include <sys/param.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include <assert.h>
#include <err.h>
#include <errno.h>
#include <netdb.h>
#include <poll.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#ifndef NO_SSL
#include "../assl/assl.h"
#endif

#include "peertask.h"
#include "peer.h"


static int	 peer_expand_cmd(struct peer *);
static char	*peer_replace_template_addr(const char *, size_t *, size_t,
		    const struct sockaddr_storage *);
static ssize_t	 peer_read(struct peer *, void *, size_t);
static ssize_t	 peer_write(struct peer *, const void *, size_t);
static int	 peer_run_task_output(struct peer *, struct peer_task *);
static int	 peer_handle_incoming_data(struct peer *, struct peer_task *);
static int	 peer_handle_incoming(struct peer *);
static void	 peer_task_restart(struct peer_task *, int);


static const char	*peer_status_names[] = {
	"PEER_STATUS_NEW",
	"PEER_STATUS_CONNECTED",
	"PEER_STATUS_HELLO_SENT",
	"PEER_STATUS_SESSION",
	"PEER_STATUS_FAILED",
	"PEER_STATUS_UNVERIFIED",
	"unknown"
};


const char *
peer_status_name(const struct peer *peer) {
	if (peer->status < PEER_STATUS_NEW || peer->status >= PEER_STATUS_COUNT)
		return peer_status_names[PEER_STATUS_COUNT];
	return peer_status_names[peer->status];
}

struct peer *
peer_new(enum peer_connect_mode mode, const char * const *argv) {
	struct peer		 *peer;
	int			  i, oerrno;
	const char * const	 *cp;
	char			**p;

	if (mode == PEER_CONN_TRANSPORT && (argv == NULL || argv[0] == NULL)) {
		errno = EINVAL;
		return NULL;
	}

	peer = malloc(sizeof(struct peer));
	if (peer == NULL)
		return NULL;
	bzero(peer, sizeof(struct peer));
	for (i = 0; i < PEER_QUEUE_COUNT; i++)
		SIMPLEQ_INIT(&peer->tasks[i]);
	switch (mode) {
	case PEER_CONN_PLAIN:
		peer->conn.fd = -1;
		break;

	case PEER_CONN_TRANSPORT:
		for (i = 0, cp = argv; *cp != NULL; i++, cp++);
		peer->trans_cmd = calloc(i + 1, sizeof(*p));
		if (peer->trans_cmd == NULL)
			goto err;
		for (i = 0, cp = argv; *cp != NULL; i++, cp++) {
			peer->trans_cmd[i] = strdup(*cp);
			if (peer->trans_cmd[i] == NULL)
				goto err;
		}
		break;

#ifndef NO_SSL
	case PEER_CONN_SSL:
		peer->flags |= PEER_FLAG_WITH_SSL;
		break;
#endif

	default:
		DPRINTFX(0, "Unknown peer connection mode %d", (int)mode);
		abort();	/* XXX */
	}
	return peer;

err:
	oerrno = errno;
	if (peer != NULL) {
		if (peer->trans_cmd != NULL) {
			for (p = peer->trans_cmd; *p != NULL; p++)
				free(p);
			free(peer->trans_cmd);
		}
		free(peer);
	}
	errno = oerrno;
	return NULL;
}

void
peer_free(struct peer *peer) {
	struct	peer_task *task;
	int	i;

	(void)peer_close(peer);
	for (i = 0; i < PEER_QUEUE_COUNT; i++)
		while (!SIMPLEQ_EMPTY(&peer->tasks[i])) {
			task = SIMPLEQ_FIRST(&peer->tasks[i]);
			SIMPLEQ_REMOVE_HEAD(&peer->tasks[i], entry);
			peer_free_task_resources(task);
			free(task);
		}
	free(peer);
}

/*
 * Replace some printf-like macros. Do not bother about saving old strings as
 * this routine is called between fork() and exec() anyway.
 */
int
peer_expand_cmd(struct peer *peer) {
	size_t	  len, pos;
	int	  marker;
	char	**pc, *p, *buf;

	for (pc = peer->trans_cmd; *pc != NULL; pc++) {
		len = strlen(*pc);
		for (marker = 0, p = *pc; *p != '\0'; p++) {
			if (*p == '%') {
				marker = !marker;
				if (!marker) {
					memmove(p, p + 1, len - (p - *pc));
					p--;
				}
				continue;
			}
			if (!marker)
				continue;

			switch (*p) {
			case 'a':
				pos = (p - *pc) - 1;    /* Point at marker */
				buf = peer_replace_template_addr(*pc, &pos, 2,
				    &peer->addr);
				if (buf == NULL) {
					DPRINTF(1, "peer_expand_cmd: "
					    "peer_replace_template_addr");
					return -1;
				}
				p = buf + pos;
				free(*pc);
				*pc = buf;
				break;

			default:
				DPRINTFX(1, "peer_expand_cmd: illegal "
				    "command character %c", (int)*p);
			}
		}
	}
	return 0;
}

/*
 * Replace template ("%a") at the specified position in the string given with
 * textual representation of socket address. Return new string, or NULL on
 * error (set errno accordingly in the latter case).
 *
 * Parameters:
 *         s - Current parameter value
 *       pos - Pointer to the offset in "s" where substitution should happen.
 *             On success the value is set to offset after the end of string
 *             substituted with.
 *   skiplen - Number of characters to skip (length of string being
 *             substituted by)
 *        ss - Address to convert.
 *
 * On success, pos set to the new position (after address).
 */
char *
peer_replace_template_addr(const char *s, size_t *pos, size_t skiplen,
    const struct sockaddr_storage *ss) {
	struct	 sockaddr_un *sun = NULL;
	size_t	 len, bufsz, npos;
	int	 oerrno;
	char	*buf;

	buf = NULL;	/* Play safe */
	switch (ss->ss_family) {
	case AF_UNIX:
		sun = (struct sockaddr_un *)ss;
		bufsz = sizeof(sun->sun_path) + 1;
		break;

	case AF_INET:
		bufsz = sizeof("255.255.255.255");
		break;

	case AF_INET6:
		/* Idea from OpenBSD src/lib/libc/net/inet_ntop.c */
		bufsz = sizeof("1111:2222:3333:4444:5555:6666:7777:8888:"
		    "255.255.255.255");
		break;

	default:
		errno = EAFNOSUPPORT;
		return NULL;
	}
	len = strlen(s);
	if (len < skiplen || *pos < len - skiplen) {
		errno = EINVAL;
		return NULL;
	}
	bufsz += len - skiplen;
	buf = calloc(bufsz, sizeof(char));
	if (buf == NULL)
		return NULL;
	memcpy(buf, s, *pos);
	if (ss->ss_family == AF_UNIX)
		/* '\0' is guaranteed by calloc() */
		strncpy(buf + *pos, sun->sun_path, sizeof(sun->sun_path));
	else
		if (inet_ntop(ss->ss_family, &((struct sockaddr *)ss)->sa_data,
		    buf + *pos, bufsz - *pos) == NULL)
			goto err;
	npos = *pos + strlen(buf + *pos);
	/* '1' comes for trailing '\0' */
	memcpy(buf + npos, s + *pos + skiplen, len - (*pos + skiplen) + 1);
	*pos = npos;
	return buf;

err:
	oerrno = errno;
	free(buf);
	errno = oerrno;
	return NULL;
}

/*
 * Make an SSL/plaintext connection to peer. As simple as it can be.
 */
int
peer_connect(struct peer *peer) {
#ifndef NO_SSL
	struct	sockaddr *sa;
	struct	assl_context *ctx;
#endif
	int	fd, oerrno, pid, pipe_chld[2];

	if (peer_connected(peer)) {
		errno = EISCONN;
		return -1;
	}
	assert(peer->trans_cmd == NULL || !peer_with_ssl(peer));

#ifndef NO_SSL
	if (peer_with_ssl(peer)) {
		ctx = assl_alloc_context(ASSL_M_TLSV1_CLIENT, 0);
		if (ctx == NULL) {
			DPRINTF(0, "peer_connect: assl_alloc_context");
			goto sslerr;
		}
		if (assl_load_file_certs(ctx,
		    peer->ssl_paths->ca_crt,
		    peer->ssl_paths->my_crt,
		    peer->ssl_paths->my_key)) {
			DPRINTF(0, "peer_connect: assl_load_file_certs");
			goto sslerr;
		}
		sa = (struct sockaddr *)&peer->addr;
		if (assl_connect_sa(ctx, sa, peer->addrlen,
		   peer->af, 0, ASSL_F_NONBLOCK)) {
			DPRINTF(2, "peer_connect: assl_connect_sa");
			goto sslerr;
		}
		peer->conn.sslctx = ctx;
	} else
#endif
	if (peer->trans_cmd != NULL) {
		pipe_chld[0] = pipe_chld[1] = -1;
		if (pipe(pipe_chld) == -1) {
			DPRINTF(0, "peer_connect: pipe");
			return -1;
		}
		pid = fork();
		if (pid == -1) {
			DPRINTF(0, "peer_connect: fork");
			goto transerr;
		}
		if (pid == 0) {
			(void)close(pipe_chld[0]);
			if (dup2(pipe_chld[1], STDOUT_FILENO) == -1) {
				DPRINTF(0, "peer_connect[child]: dup2");
				_exit(EXIT_FAILURE);
			}
			(void)close(pipe_chld[1]);
			if (peer_expand_cmd(peer) == -1) {
				DPRINTF(0, "peer_expand_cmd");
				_exit(EXIT_FAILURE);
			}
			execvp(peer->trans_cmd[0], peer->trans_cmd);
			DPRINTF(0, "peer_connect[child]: execvp");
			_exit(EXIT_FAILURE);
		}
		(void)close(pipe_chld[1]);
		peer->conn.fd = pipe_chld[0];
	} else {
		fd = socket(peer->af, SOCK_STREAM, 0);
		if (fd == -1) {
			DPRINTF(0, "peer_connect: socket");
			return -1;
		}
		if (connect(fd, (struct sockaddr *)&peer->addr, peer->addrlen)){
			DPRINTF(2, "peer_connect: connect");
			goto err;
		}
		peer->conn.fd = fd;
	}

	peer->status = PEER_STATUS_CONNECTED;
	return 0;

err:
	oerrno = errno;
	(void)close(fd);
	errno = oerrno;
	return -1;

transerr:
	oerrno = errno;
	(void)close(pipe_chld[0]);
	(void)close(pipe_chld[1]);
	errno = oerrno;
	return -1;

#ifndef NO_SSL
sslerr:
	oerrno = errno;
	(void)assl_close(ctx);
	errno = oerrno;
	return -1;
#endif
}

/*
 * Accepts SSL/plaintext connection from peer.
 */
int
peer_accept(struct peer *peer, int sock) {
#ifndef NO_SSL
	struct		assl_context *ctx;
#endif
	socklen_t	len;
	int		fd;
#ifndef NO_SSL
	int		oerrno;
#endif

#ifndef NO_SSL
	if (peer_with_ssl(peer)) {
		if (peer->conn.sslctx != NULL) {
			errno = EISCONN;
			return -1;
		}
	} else {
#endif
		if (peer->conn.fd != -1) {
			errno = EISCONN;
			return -1;
		}
#ifndef NO_SSL
	}
#endif

	fd = accept(sock, (struct sockaddr *)&peer->addr, &len);
	if (fd == -1)
		return -1;
	peer->addrlen = len;
#ifndef __linux__
	peer->addr.ss_len = len;
#endif
	peer->af = peer->addr.ss_family;

#ifndef NO_SSL
	if (peer_with_ssl(peer)) {
		ctx = assl_alloc_context(ASSL_M_TLSV1_SERVER, 0);
		if (peer->conn.sslctx == NULL) {
			DPRINTF(0, "peer_accept: assl_alloc_context");
			goto sslerr;
		}
		if (assl_load_file_certs(peer->conn.sslctx,
		    peer->ssl_paths->ca_crt,
		    peer->ssl_paths->my_crt,
		    peer->ssl_paths->my_key)) {
			DPRINTF(0, "peer_accept: assl_load_file_certs");
			goto sslerr;
		}
		if (assl_accept(ctx, fd)) {
			DPRINTF(2, "peer_accept: assl_accept");
			goto sslerr;
		}
	} else
#endif
		peer->conn.fd = fd;
	peer->status = PEER_STATUS_CONNECTED;
	return 0;

#ifndef NO_SSL
sslerr:
	oerrno = errno;
	(void)assl_close(ctx);
	(void)close(fd);
	errno = oerrno;
	return -1;
#endif
}

/*
 * Attempt to start session by sending HELLO message. Peer will gain
 * HELLO_SENT status on success.
 * 
 * Return -1 on error (and set errno), 0 on success.
 */
int
peer_start_session(struct peer *peer, uint32_t session, const char *myname) {
	struct	peer_task *task;
	struct	peer_data_buffer *buf;
	struct	peer_msg_hello *hmsg;
	size_t	len;

	if (peer->status != PEER_STATUS_CONNECTED) {
		errno = ENOTTY;
		return -1;
	}

	buf = NULL;
	task = peer_new_task(1, NULL);
	if (task == NULL)
		goto err;
	len = strlen(myname);
	buf = peer_new_data_buffer(sizeof(struct peer_msg_hello) + len, NULL);
	if (buf == NULL)
		goto err;
	/* XXX: possible pointer alignment problem */
	hmsg = (struct peer_msg_hello *)buf->data;
	hmsg->version = htonl(PEER_VERSION);
	memcpy(hmsg->peer_name, myname, len);
	peer_init_task(task, PEER_CMD_HELLO);
	peer_add_task_data(task, buf, -1);
	if (peer->session < session)
		peer->session = session;
	if (peer_add_task(peer, task) == -1)
		goto err;
	return 0;

err:
	free(buf);
	free(task);
	return -1;
}

/*
 * Following three functions just call read(2)/write(2)/close(2) or their assl_*
 * counterparts, depending on using PEER_FLAG_WITH_SSL.
 */
ssize_t
peer_read(struct peer *peer, void *buf, size_t nbytes) {
#ifndef NO_SSL
	if (peer_with_ssl(peer))
		return assl_read(peer->conn.sslctx, buf, nbytes);
	else
#endif
		return read(peer->conn.fd, buf, nbytes);
}

ssize_t
peer_write(struct peer *peer, const void *buf, size_t nbytes) {
#ifndef NO_SSL
	if (peer_with_ssl(peer))
		return assl_write(peer->conn.sslctx, buf, nbytes);
	else
#endif
		return write(peer->conn.fd, buf, nbytes);
}

int
peer_close(struct peer *peer) {
	struct	peer_task *task;
	int	rv;

#ifndef NO_SSL
	if (peer_with_ssl(peer)) {
		rv = assl_close(peer->conn.sslctx);
		peer->conn.sslctx = NULL;
	} else {
#endif
		rv = close(peer->conn.fd);
		peer->conn.fd = -1;
#ifndef NO_SSL
	}
#endif

	if (peer->incoming) {
		peer_free_task(peer->incoming, 1);
		peer->incoming = NULL;
	}
	SIMPLEQ_FOREACH(task, &peer->tasks[PEER_QUEUE_OUTPUT], entry)
		if (task->phase != PEER_TASK_PHASE_START &&
		    task->phase != PEER_TASK_PHASE_DONE)
			peer_task_restart(task, PEER_QUEUE_OUTPUT);
	peer->status = PEER_STATUS_NEW;
	return rv;
}

/*
 * This function does not look at status but at the corresponding descriptor.
 */
int
peer_is_closed(struct peer *peer) {
#ifndef NO_SSL
	if (peer_with_ssl(peer))
		return (peer->conn.sslctx == NULL);
	else
#endif
		return (peer->conn.fd == -1);
}

/*
 * Check and reset HAS_COMPLETED TASKS flag.
 */
int
peer_has_completed_tasks(struct peer *peer) {
	int rv;

	rv = (peer->flags & PEER_FLAG_HAS_COMPLETED_TASKS);
	peer->flags &= ~PEER_FLAG_HAS_COMPLETED_TASKS;
	return rv;
}
/*
 * Return -1 and set errno on error other that EAGAIN or EINTR. If no error
 * occured, or system routine returned EAGAIN or EINTR, return 0.
 *
 * Notice that it could be run not only on tasks put in queue.
 */
int
peer_run_task_output(struct peer *peer, struct peer_task *task) {
	ssize_t	len, n, bufprogress;
	int	i;
	const	char *p;

	if (!peer_connected(peer)) {
		errno = EBADF;
		return -1;
	}

	switch (task->phase) {
	case PEER_TASK_PHASE_START:
		task->phase = PEER_TASK_PHASE_CMD;
		task->phase_progress = 0;
		/* FALLTHROUGH */

	case PEER_TASK_PHASE_CMD:
		task->error = 0;
		p = (char*)&task->cmd;
		n = peer_write(peer, p + task->phase_progress,
		    sizeof(struct peer_msg_hdr) - task->phase_progress);
		if (n == -1 && errno != EAGAIN)
			goto err;
		task->phase_progress += n;
		if (task->phase_progress == 0)
			task->phase = PEER_TASK_PHASE_START;
		if (task->phase_progress < sizeof(struct peer_msg_hdr))
			return 0;

		task->phase_progress = 0;
		switch (task_cmd_code(task)) {
		case PEER_CMD_OK:
		case PEER_CMD_ERROR:
			/* no payload and waiting here */
			assert(task->cmd.pmh_bufcount == 0);
			task->phase = PEER_TASK_PHASE_DONE;
			goto task_done;
		}
		if (task->cmd.pmh_bufcount == 0) {
			task->phase = PEER_TASK_PHASE_WAIT;
			goto task_wait;
		}
		task->phase = PEER_TASK_PHASE_DATA;
		/* FALLTHROUGH */

	case PEER_TASK_PHASE_DATA:
		task->error = 0;
		bufprogress = task->phase_progress;
		for (i = 0; i < ntohs(task->cmd.pmh_bufcount); i++) {
			len = sizeof(task->databuf[i]->len) +
			    task->databuf[i]->len;
			if (bufprogress > len) {
				bufprogress -= len;
				continue;
			}
			p = (const char *)&task->databuf[i]->len;
			task->databuf[i]->len = htonl(task->databuf[i]->len);
			n = peer_write(peer, p + bufprogress,
			    len - bufprogress);
			task->databuf[i]->len = ntohl(task->databuf[i]->len);
			if (n == -1 && errno != EAGAIN)
				goto err;
			bufprogress += n;
			task->phase_progress += n;
			if (bufprogress < len)
				return 0;
		}
		task->phase = PEER_TASK_PHASE_WAIT;
		goto task_wait;

	case PEER_TASK_PHASE_WAIT:
task_wait:
		/*
		 * When OK message will arrive, peer_handle_incoming() will
		 * automatically adjust status of the current task.
		 *
		 * Timeouts should be implemented here.
		 */
		return 0;

	case PEER_TASK_PHASE_DONE:
task_done:
		peer->flags |= PEER_FLAG_HAS_COMPLETED_TASKS;
		if (is_error_task(task)) {
			peer->fail_reason = errno = EPIPE;
			peer->status = PEER_STATUS_FAILED;
			break;
		}
	}
	return 0;

err:
	peer->fail_reason = task->error = errno;
	peer->status = PEER_STATUS_FAILED;
	peer_task_restart(task, PEER_QUEUE_OUTPUT);
	errno = peer->fail_reason;
	return -1;
}

/*
 * Side-effect: always sets PEER_FLAG_HAS_COMPLETED_TASKS if there are any
 * completed tasks in queue, failed or succeeded.
 */
void
peer_run_queues(struct peer *peer) {
	struct	peer_task *task, *tt;
	time_t	now;

	now = time(NULL);
	/* Process output first to free as much resources as possible */
	for (task = SIMPLEQ_FIRST(&peer->tasks[PEER_QUEUE_OUTPUT]); task;) {
		if (task->phase == PEER_TASK_PHASE_DONE) {
			peer->flags |= PEER_FLAG_HAS_COMPLETED_TASKS;
			tt = SIMPLEQ_NEXT(task, entry);
			if (is_hello_task(task)) {
				/* Catch on session establishment */
				peer->status = PEER_STATUS_SESSION;
				peer_free_task(task, 1);
			} else if (is_ok_task(task))
				/* Eat all OKs ourself as they do not matter */
				peer_free_task(task, 1);
			task = tt;
			continue;
		}
		if (task->started - now > PEER_TASK_TIMEOUT) {
			peer->fail_reason = task->error = ETIMEDOUT;
			peer->status = PEER_STATUS_FAILED;
			task->phase = PEER_TASK_PHASE_DONE;
			return;
		}
		if (peer_run_task_output(peer, task)) {
			if (errno == EAGAIN || errno == EINTR)
				goto handle_input;
			return;
		}
		task = SIMPLEQ_NEXT(task, entry);
	}

handle_input:
	while (peer_handle_incoming(peer) == 0);
	return;
}

/*
 * Read all message data buffers.
 *
 * Returns 0 if all data buffers were read completely, otherwise return -1
 * with corresponding errno set.
 */
int
peer_handle_incoming_data(struct peer *peer, struct peer_task *task) {
	ssize_t	 len, n, bufprogress;
	int	 i;
	char	*p;

	bufprogress = task->phase_progress;
	for (i = 0; i < ntohs(task->cmd.pmh_bufcount); i++) {
		if (i >= MAX_TASK_BUFFERS) {
			errno = ENOBUFS;
			return -1;
		}

		/* at first we need get the length of buffer */
		len = sizeof(task->_buf_len);
		if (bufprogress >= len)
			goto len_read;
		p = (char *)&task->_buf_len;
		n = peer_read(peer, p + bufprogress, len - bufprogress);
		if (n == -1)
			return -1;
		bufprogress += n;
		task->phase_progress += n;
		if (bufprogress < len) {
			errno = EAGAIN;
			return -1;
		}
		task->_buf_len = ntohl(task->_buf_len);

len_read:
		if (task->databuf[i] == NULL) {
			task->databuf[i] = peer_new_data_buffer(task->_buf_len,
			    NULL);
			if (task->databuf[i] == NULL)
				return -1;
		} else if (bufprogress >= task->databuf[i]->len) {
			bufprogress -= task->databuf[i]->len;
			continue;
		}
		if (task->databuf[i]->len > MAX_PEER_DATA) {
			DPRINTFX(1, "peer tried to send data of %lu length",
			    (unsigned long)task->databuf[i]->len);
			errno = ENOBUFS;
			return -1;
		}
		p = (char *)task->databuf[i]->data;
		n = peer_read(peer, p + bufprogress,
		    task->databuf[i]->len - bufprogress);
		if (n == -1)
			return -1;
		task->phase_progress += n;
		bufprogress += n;
		if (bufprogress < task->databuf[i]->len) {
			errno = EAGAIN;
			return -1;
		}
	}
	return 0;
}

/*
 * Read up to one message from peer into peer->incoming.
 *
 * Returns 0 if message read completely, otherwise return -1.
 */
int
peer_handle_incoming(struct peer *peer) {
	struct	 peer_task *task;
	ssize_t	 n;
	int	 code;
	char	*p;

	if (peer->qlength[PEER_QUEUE_INPUT] >= PEER_MAX_QUEUE_LENGTH) {
		peer->fail_reason = errno = ENOBUFS;
		peer->status = PEER_STATUS_FAILED;
		return -1;
	}

	if (peer->incoming == NULL) {
		peer->incoming = malloc(sizeof(struct peer_task));
		if (peer->incoming == NULL)
			return -1;
		bzero(peer->incoming, sizeof(struct peer_task));
	}

	switch (peer->incoming->phase) {
	case PEER_TASK_PHASE_START:
		peer->incoming->phase = PEER_TASK_PHASE_CMD;
		peer->incoming->phase_progress = 0;
		/* FALLTHROUGH */

	case PEER_TASK_PHASE_CMD:
		p = (char *)&peer->incoming->cmd;
		n = peer_read(peer, p + peer->incoming->phase_progress,
		    sizeof(struct peer_msg_hdr) -
		    peer->incoming->phase_progress);
		if (n == -1 && errno != EAGAIN && errno != EINTR)
			goto err;
		peer->incoming->phase_progress += n;
		if (peer->incoming->phase_progress == 0)
			peer->incoming->phase = PEER_TASK_PHASE_START;
		if (peer->incoming->phase_progress < sizeof(struct peer_msg_hdr))
			return 0;
		/* XXX: Should add some sanity checks */
		peer->incoming->phase_progress = 0;
		code = ntohs(peer->incoming->cmd.pmh_code);
		if (code <= PEER_CMD_UNKNOWN || code >= PEER_CMD_COUNT) {
			DPRINTFX(1, "peer_handle_incoming: "
			    "unknown command code %d", code);
			errno = EINVAL;
			goto err;
		}
		switch (code) {
		case PEER_CMD_OK:
		case PEER_CMD_ERROR:
			if (peer->incoming->cmd.pmh_bufcount != 0) {
				DPRINTFX(1, "peer_handle_incoming: "
				    "no data awaited, abort transfers");
				errno = EINVAL;    /* XXX: better code? */
				goto err;
			}
		}
		peer->incoming->phase = PEER_TASK_PHASE_DATA;
		/* FALLTHROUGH */

	case PEER_TASK_PHASE_DATA:
		if (peer_handle_incoming_data(peer, peer->incoming)) {
			if (errno != EAGAIN && errno != EINTR)
				goto err;
			return 0;
		}
		peer->incoming->phase = PEER_TASK_PHASE_DONE;
		/* FALLTHROUGH */

	case PEER_TASK_PHASE_DONE:
		if (is_ok_task(peer->incoming)) {
			/*
	 		 * Do not save in the input queue but instead mark
	 		 * the corresponding task in output queue as done.
	 		 */
			SIMPLEQ_FOREACH(task, &peer->tasks[PEER_QUEUE_OUTPUT],
			    entry)
				if (task->cmd.pmh_seq == peer->incoming->cmd.pmh_seq &&
				    task->cmd.pmh_session ==
				    peer->incoming->cmd.pmh_session) {
					task->phase = PEER_TASK_PHASE_DONE;
					peer_free_task(peer->incoming, 1);
					goto task_finish;
				}
			DPRINTFX(1, "unknown task ID in OK message: %lu:%lu",
			    (unsigned long)ntohl(peer->incoming->cmd.pmh_session),
			    (unsigned long)ntohl(peer->incoming->cmd.pmh_seq));
			peer_free_task_resources(peer->incoming);
			peer_init_error_task(peer->incoming, ESRCH);
			peer_add_task(peer, peer->incoming);
			goto task_finish;
		}
		SIMPLEQ_INSERT_TAIL(&peer->tasks[PEER_QUEUE_INPUT],
		    peer->incoming, entry);
		peer->qlength[PEER_QUEUE_INPUT]++;
task_finish:
		peer->incoming = NULL;
		peer->flags |= PEER_FLAG_HAS_COMPLETED_TASKS;
	}
	return 0;

err:
	peer->fail_reason = errno;
	peer->status = PEER_STATUS_FAILED;
	peer_free_task_resources(peer->incoming);
	free(peer->incoming);
	peer->incoming = NULL;
	return -1;
}

/*
 * Add task to peer's output queue.
 * 
 * Can fail (return -1) with ENOBUFS if message queue is full, ENOTTY if message
 * is inapporiate. Always succeeds (return 0) for error and hello messages, they
 * can go over quota.
 */
int
peer_add_task(struct peer *peer, struct peer_task *task) {
	struct	peer_task_queue *queue;
	struct	peer_task *t, *prev;

	queue = &peer->tasks[PEER_QUEUE_OUTPUT];
	DPTASK(3, task, "peer_add_task");

	if (peer->qlength[PEER_QUEUE_OUTPUT] >= PEER_MAX_QUEUE_LENGTH &&
	    !is_error_task(task) && !is_hello_task(task)) {
		errno = ENOBUFS;
		return -1;
	}
	if (is_hello_task(task) && peer->status != PEER_STATUS_CONNECTED) {
		errno = ENOTTY;
		return -1;
	}

	if (is_hello_task(task)) {
		if (peer->session != 0)
			peer->session++;
		do {
			peer->lastseq = arc4random();
		} while (peer->lastseq == 0xFFFFFFFFL);
	} else if (!is_error_task(task)) {
		if (peer->lastseq == 0xFFFFFFFFL) {
			peer->session++;
			peer->lastseq = 1;
		}
	}
	task->cmd.pmh_session = htonl(peer->session);
	task->cmd.pmh_seq = htonl(peer->lastseq);
	task->started = time(NULL);
	if (is_hello_task(task)) {
		/* Anyway session should be established before any other data */
		SIMPLEQ_INSERT_HEAD(queue, task, entry);
		goto end;
	}
	if (is_error_task(task) && !SIMPLEQ_EMPTY(queue)) {
		t = SIMPLEQ_FIRST(queue);
		if (t->phase == PEER_TASK_PHASE_START) {
			SIMPLEQ_INSERT_HEAD(queue, task, entry);
			goto end;
		}
		for (prev = t; (t = SIMPLEQ_NEXT(t, entry)) != NULL; prev = t)
			if (t->phase == PEER_TASK_PHASE_START) {
				SIMPLEQ_INSERT_AFTER(queue, prev, task, entry);
				goto end;
			}
	}
	SIMPLEQ_INSERT_TAIL(queue, task, entry);

end:
	peer->qlength[PEER_QUEUE_OUTPUT]++;
	return 0;
}

/*
 * Remove a completed task from queue. The caller should call peer_free_task()
 * on the pointer after he's done with it.
 *
 * If no completed tasks available - NULL is returned with errno set to ENOENT.
 * If illegal queue ID supplied - NULL is returned with errno set to EINVAL.
 */
struct peer_task *
peer_pick_done_task(struct peer *peer, int qid) {
	struct	peer_task_queue *queue;
	struct	peer_task *task;

	if (qid != PEER_QUEUE_INPUT && qid != PEER_QUEUE_OUTPUT) {
		errno = EINVAL;
		goto err;
	}
	queue = &peer->tasks[qid];

	if (SIMPLEQ_EMPTY(queue)) {
		errno = ENOENT;
		goto err;
	}
	task = SIMPLEQ_FIRST(queue);
	if (task->phase != PEER_TASK_PHASE_DONE) {
		errno = ENOENT;
		goto err;
	}
	SIMPLEQ_REMOVE_HEAD(queue, entry);
	peer->qlength[qid]--;
	DPTASK(3, task, "peer_pick_done_task");
	return task;

err:
	DPRINTF(3, "peer_pick_done_task");
	return NULL;
}

/*
 * Clean up task object, for retrying send/receive later.
 *
 * If run on task from input queue, free occupied resources.
 */
void
peer_task_restart(struct peer_task *task, int qid) {
	if (qid == PEER_QUEUE_INPUT)
		peer_free_task_resources(task);
	task->phase_progress = 0;
	task->phase = PEER_TASK_PHASE_START;
	/*
	 * Do not touch "error", it will be reset in the first task read or
	 * write operation.
	 */
}

/*
 * Try to send tasks from queue to peers, not more than one task peer. Tasks
 * only put in queue of those peers who has the session established.
 */
void
peer_send_tasks_to_any(struct peer_task_queue *queue, struct peer_list *peers) {
	struct	peer_task *task;
	struct	peer *peer;

	peer = TAILQ_FIRST(peers);
	if (peer == NULL)
		/* Do not loop extra time */
		return;
	while(!SIMPLEQ_EMPTY(queue)) {
		task = SIMPLEQ_FIRST(queue);
		for (; peer != NULL && peer_session_established(peer);
		    peer = TAILQ_NEXT(peer, entry));
		if (peer == NULL)
			return;
		SIMPLEQ_REMOVE_HEAD(queue, entry);
		peer_add_task(peer, task);
		DPTASK(2, task, "Task taken out of unsent queue\n");
	}
}

/*
 * Implement common pattern of tasks queues handling. Note that order of
 * handlers is meaningful because the first match succeeds.
 */
int
peer_handle(struct peer *peer, struct peer_task_handler *hs, size_t count) {
	struct	peer_task *task;
	size_t	i, qid;
	int	rv;

	rv = 0;
	if (!(peer->pollptr->revents & (POLLOUT|POLLIN)))
		goto end;
	peer_run_queues(peer);
	if (!peer_has_completed_tasks(peer))
		goto end;
	for (qid = 0; qid < PEER_QUEUE_COUNT; qid++)
		while ((task = peer_pick_done_task(peer, qid))) {
		for (i = 0; i < count; i++) {
			if (hs[i].queue != qid)
				continue;
			if (hs[i].cmdcode != PEER_CMD_UNKNOWN &&
			    ntohs(task->cmd.pmh_code) != hs[i].cmdcode)
				continue;
			if (peer_task_failed(task)) {
				if (!(hs[i].flags & PEER_TH_F_FAIL))
					continue;
			} else {
				if (!(hs[i].flags & PEER_TH_F_SUCCESS))
					continue;
			}
			switch ((*hs[i].func)(peer, task)) {
				case PEER_TH_CB_ERR:
					SIMPLEQ_INSERT_HEAD(&peer->tasks[qid],
					    task, entry);
					peer->flags |=
					    PEER_FLAG_HAS_COMPLETED_TASKS;
					goto err;

				case PEER_TH_CB_RESTART:
					assert(qid == PEER_QUEUE_OUTPUT);
					peer_task_restart(task, qid);
					if (peer_add_task(peer, task) == -1)
						goto err;
					break;

				case PEER_TH_CB_FREE:
					peer_free_task(task, 1);
					/* FALLTHROUGH */

				case PEER_TH_CB_CARED:
					break;
			}
			break;
		}
		if (i == count)
			/* XXX: "no handler found" is not an error now */
			peer_free_task(task, 1);
	}
end:
	peer->pollptr->revents &= ~(POLLOUT|POLLIN);
	if (peer->pollptr->revents)
		peer->status = PEER_STATUS_FAILED;
	return rv;

err:
	peer->status = PEER_STATUS_FAILED;
	return -1;
}

/*
 * Try to (re)establish connections with peers. If addrs != NULL then also check
 * if peer addresses list changed, drop extra connections and perform new where
 * needed.
 *
 * Return 0 if nothing changed, or number of dropped+established connections
 * otherwise.
 */
#ifdef NO_SSL
int
peer_update_connections(struct peer_list *peers, struct addrinfo *addrs,
    enum peer_connect_mode mode, int *sesscount, const char * const *trans_cmd,
    void *unused) {
#else
int
peer_update_connections(struct peer_list *peers, struct addrinfo *addrs,
    enum peer_connect_mode mode, int *sesscount, const char * const *trans_cmd,
    struct ssl_paths *ssl_paths) {
#endif
	struct	addrinfo *ai;
	struct	peer *peer, *peer2;
	struct	sockaddr *sa;
	int	changes, found, nsessions;
	char	buf[MAXPEERADDRLEN];

	changes = 0;

	if (addrs != NULL) {
		TAILQ_FOREACH(peer, peers, entry)
			peer->status = PEER_STATUS_UNVERIFIED;

		for (ai = addrs; ai != NULL; ai = ai->ai_next) {
			found = 0;
			TAILQ_FOREACH(peer, peers, entry) {
				sa = (struct sockaddr *)&peer->addr;
				if (peer->addrlen == ai->ai_addrlen &&
				memcmp(sa, ai->ai_addr, peer->addrlen) == 0) {
					found = 1;
					break;
				}
			}
			if (!found) {
				/* XXX: incompatible pointer type??? */
				peer = peer_new(mode, trans_cmd);
				peer->af = ai->ai_family;
				sa = (struct sockaddr *)&peer->addr;
				memcpy(&sa->sa_data, ai->ai_addr,
				    ai->ai_addrlen);
#ifndef NO_SSL
				peer->ssl_paths = ssl_paths;
#endif
				peer->status = PEER_STATUS_NEW;
				TAILQ_INSERT_TAIL(peers, peer, entry);
				/* Do not try to connect here */
			}
		}
	}

	nsessions = 0;
	for (peer = TAILQ_FIRST(peers); peer != NULL;) {
		switch (peer->status) {
		case PEER_STATUS_SESSION:
			nsessions++;
			break;

		case PEER_STATUS_UNVERIFIED:
			/* Not in (DNS) address list anymore */
			peer2 = TAILQ_NEXT(peer, entry);
			TAILQ_REMOVE(peers, peer, entry);
			peer_free(peer);
			peer = peer2;
			changes++;
			break;

		case PEER_STATUS_FAILED:
			if (!inet_ntop(peer->af, &peer->addr, buf, sizeof(buf)))
				strlcpy(buf, "[inet_ntop failed]", sizeof(buf));
			DPRINTFX(2, "peer %s failed: %s", buf,
			    strerror(peer->fail_reason));
			assert(!peer_is_closed(peer));
			(void)peer_close(peer);
			changes++;
			peer->status = PEER_STATUS_NEW;
			/* FALLTHROUGH */

		case PEER_STATUS_NEW:
			/* Not connected yet, retries here are infinite */
			assert(peer_is_closed(peer));
			if (peer_connect(peer) == 0) {
				/*
				 * Increment only on successful connection
				 * as we're using this to update an fd pooling
				 * array.
				 */
				changes++;
			}
			/* FALLTHROUGH */

		default:
			peer = TAILQ_NEXT(peer, entry);
		}
	}
	if (sesscount)
		*sesscount = nsessions;
	return changes;
}
